package com.foreveross.taskservice.taskcache.redis;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.UUID;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.ArrayUtils;
import org.springframework.stereotype.Service;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

import com.foreveross.taskservice.common.CacheDataSpace;
import com.foreveross.taskservice.common.TaskStatusEnum;
import com.foreveross.taskservice.common.model.AbstractTaskModel;
import com.foreveross.taskservice.infra.cfg.PropertyUtils;
import com.foreveross.taskservice.infra.cfg.taskservice.TaskServicePropertiesLoader;
import com.foreveross.taskservice.taskcache.ITaskCacheQuery;
import com.foreveross.taskservice.taskcache.QueryCondition;
import com.foreveross.taskservice.taskcache.QueryResult;
import com.foreveross.taskservice.taskcache.redis.core.TaskRedisCallback;
import com.foreveross.taskservice.taskcache.redis.util.KeyUtil;
import com.foreveross.taskservice.taskcache.redis.util.RedisSerializeUtil;
import com.foreveross.taskservice.taskcache.redis.util.TaskScopeUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
@Service
public class RedisTaskCacheQueryImpl extends BaseRedisTaskCache implements ITaskCacheQuery{
	@Override
	public long queryTaskSize(CacheDataSpace d, String channelCode,
			TaskStatusEnum status) throws Exception {
		Map<String, Map<TaskStatusEnum, Long>> map=queryTaskSize(d, new String[]{channelCode}, new TaskStatusEnum[]{status});
		//return map.get(channelCode).get(status);
		return map.size()>0?map.get(channelCode).size():0l;
	}

	//新模型之后方法
	@Override
	public List<AbstractTaskModel> queryTaskDetailInfo(final String channelCode,
			final TaskStatusEnum status, int offset, int count) throws Exception {
		return excute(new TaskRedisCallback<List<AbstractTaskModel>>() {

			@Override
			public List<AbstractTaskModel> doInRedis(Jedis j) throws Exception {
				String key=KeyUtil.getKey(channelCode, status.getStatus()+"");
				
				
				
				
				
				return null;
			}
			
		});
	}
	
	
	@Override
	public List<AbstractTaskModel> queryTaskDetailInfo(final CacheDataSpace d,
			final String channelCode, final TaskStatusEnum status, final int offset, final int count)
			throws Exception {
		return excute(new TaskRedisCallback<List<AbstractTaskModel>>() {

			@Override
			public List<AbstractTaskModel> doInRedis(Jedis j) throws Exception {
				Set<byte[]> ids;
				if(status==TaskStatusEnum.TASK_WATI){
					ids=j.zrevrange(buildChannelWaitQueueKeyByte(d, channelCode), offset, offset+count);
				}else if(status==TaskStatusEnum.TASK_DOING){
					ids=j.zrevrange(buildChannelDoingQueueKeyByte(d, channelCode),offset, offset+count);
				}else if(status.getStatus() <= TaskStatusEnum.TASK_BUSI_FAIL.getStatus() || 
									status.getStatus() >= TaskStatusEnum.TASK_BUSI_SUCCESS.getStatus()){
					if(status.getStatus()<=TaskStatusEnum.TASK_BUSI_FAIL.getStatus()){//这里要包含所有的业务定义类型任务
						ids=j.zrangeByScore(buildChannelFinishQueueKeyByte(d, channelCode), -1000, status.getStatus(), offset, offset+count);
					}else{
						ids=j.zrangeByScore(buildChannelFinishQueueKeyByte(d, channelCode), status.getStatus(), 1000, offset, offset+count);
					}
				}else{
					double[] states = TaskScopeUtil.getFinishQueueMinMaxScope(status.getStatus());
					ids=j.zrangeByScore(buildChannelFinishQueueKeyByte(d, channelCode), states[0], states[1], offset, count);
				}
				return Lists.newArrayList(findCacheTasks((LinkedHashSet<byte[]>) ids, true, j));
			}
		});
	}
	
	
	
	public List<AbstractTaskModel> queryTaskDetailInfo(Date d,int taskType,String channelCode,TaskStatusEnum status,int offset,int count)throws Exception{
		return queryTaskDetailInfo(new CacheDataSpace(d, taskType), channelCode, status, offset, count);
	};

	@Override
	public long queryFinishTaskSize(final CacheDataSpace d, final String channelId)throws Exception {
		return excute(new TaskRedisCallback<Long>() {

			@Override
			public Long doInRedis(Jedis j) throws Exception {
				return j.zcard(buildChannelFinishQueueKeyByte(d, channelId));
			}
		});
	}
	@Override
	public long queryFinishTaskSize(Date d,int taskType,String channelId)throws Exception {
		return queryFinishTaskSize(new CacheDataSpace(d, taskType), channelId);
	}
	

	@Override
	public Map<String, Map<TaskStatusEnum, Long>> queryTaskSize(final CacheDataSpace d,
			final String[] channelCodes, final TaskStatusEnum[] status) throws Exception {
		final List<Long> result=new ArrayList<Long>();
		excute(new TaskRedisCallback<List<Long>>() {
			@Override
			public List<Long> doInRedis(Jedis j) throws Exception {
				Pipeline pl=openPipeline(j, false);
//				LinkedHashSet<AbstractTaskModel> result = new LinkedHashSet<AbstractTaskModel>();
				for(String id:channelCodes){
					for(TaskStatusEnum s:status){
						//新版redis
						//组装成一个模糊KEY
						List<byte[]> keyLists=new ArrayList<byte[]>();
						keyLists=getKeys(id, s.getStatus()+"");
						if(keyLists.size()>0){
							result.add((long)(findCacheTasks(Sets.newLinkedHashSet(keyLists), true, j)).size());
						}
						
						/*if(s==TaskStatusEnum.TASK_WATI){
//							pl.zcard(buildChannelWaitQueueKeyByte(d, id));//TASKDETAIL:HASH
//							pl.zcard(buildChannelWaitQueueKeyByteNew(d, id));
							//result.add(RedisSerializeUtil.deserializeObject(pl.get(RedisSerializeUtil.serializeString("0000012014102700002100201410292014102973936296")),AbstractTaskModel.class));
							pl.get(RedisSerializeUtil.serializeString("0000012014102700002100201410292014102973936296"));
							pl.zcard("0000012014102700002100201410292014102973936296".getBytes());
//							pl.hget(RedisSerializeUtil.serializeString("TASKDETAIL:HASH"),RedisSerializeUtil.serializeString("0000012014102400002100*"));
						}else if(s==TaskStatusEnum.TASK_DOING){
							pl.zcard(buildChannelDoingQueueKeyByte(d,id));
						}else if(s==TaskStatusEnum.TASH_PAUSE){
							pl.zcard(buildChannelPauseQueueKeyByte(d, id));
						}else if(s.getStatus() <= TaskStatusEnum.TASK_BUSI_FAIL.getStatus() || 
								s.getStatus() >= TaskStatusEnum.TASK_BUSI_SUCCESS.getStatus()){
							if(s.getStatus()<=TaskStatusEnum.TASK_BUSI_FAIL.getStatus()){//这里要包含所有的业务定义类型任务
								pl.zcount(buildChannelFinishQueueKeyByteNew(d, id), -10000, s.getStatus());
							}else{
								pl.zcount(buildChannelFinishQueueKeyByteNew(d, id), s.getStatus(), 10000);
							}
						}else{
							double[] states = TaskScopeUtil.getFinishQueueMinMaxScope(s.getStatus());
							pl.zcount(buildChannelFinishQueueKeyByte(d, id), states[0], states[1]);
						}*/
					}
				}
				return syncAndReturnAllPipline(pl, false, Long.class, false);
			}
		});
		return splitData(result,channelCodes,status);
	}
	private <T> Map<String, Map<TaskStatusEnum, T>> splitData(List<T> data,String[] channelCodes,TaskStatusEnum[] status){
		Map<String, Map<TaskStatusEnum, T>> map=Maps.newLinkedHashMap();
		int i=0;
		if(data.size()>0){
			for(String id:channelCodes){
				if(!map.containsKey(id)){
					map.put(id,new LinkedHashMap<TaskStatusEnum,T>());
				}
				
				for(TaskStatusEnum s:status){
					map.get(id).put(s, data.get(i));
					i++;
				}	
			}
		}
		return map;
	}
	@Override
	public long queryChannelAllCrawlCount(final CacheDataSpace d, final String channelId)throws Exception {
		return excute(new TaskRedisCallback<Long>() {
			@Override
			public Long doInRedis(Jedis j) throws Exception {
				return j.incrBy(buildCountInfoKeyByte(d, channelId, TaskCacheContext.SPACE_CRAWLCOUNT), 0);
			}
		});
	}
	public long queryChannelAllCrawlCount(Date d,int taskType,String channelId)throws Exception{
		return queryChannelAllCrawlCount(new CacheDataSpace(d,taskType),channelId);
	};

	@Override
	public long[] queryChannelAllCrawlCounts(final CacheDataSpace d,final String[] channelIds) throws Exception {
		List<Long> list= excute(new TaskRedisCallback<List<Long>>() {

			@Override
			public List<Long> doInRedis(Jedis j) throws Exception {
				Pipeline pl=openPipeline(j, false);
				for(String id:channelIds){
					pl.incrBy(buildCountInfoKeyByte(d, id, TaskCacheContext.SPACE_CRAWLCOUNT), 0);
				}
				return syncAndReturnAllPipline(pl, false, Long.class, false);
			}
		});
		return ArrayUtils.toPrimitive(list.toArray(new Long[0]));
	}
	public long[] queryChannelAllCrawlCounts(Date d,int taskType,String[] channelIds)throws Exception{
		return queryChannelAllCrawlCounts(new CacheDataSpace(d,taskType),channelIds);
	};
	@Override
	public long queryChannelAllByteLength(final CacheDataSpace d, final String channelId)
			throws Exception {
		return excute(new TaskRedisCallback<Long>() {
			@Override
			public Long doInRedis(Jedis j) throws Exception {
				return j.incrBy(buildCountInfoKeyByte(d, channelId, TaskCacheContext.SPACE_BYTELENTH), 0);
			}
		});
	}
	
	public long queryChannelAllByteLength(Date d,int taskType,String channelId)throws Exception{
		return queryChannelAllByteLength(new CacheDataSpace(d, taskType),channelId);
	};

	@Override
	public long[] queryChannelAllByteLength(final CacheDataSpace d,
			final String[] channelIds) throws Exception {
		List<Long> list=excute(new TaskRedisCallback<List<Long>>() {

			@Override
			public List<Long> doInRedis(Jedis j) throws Exception {
				Pipeline pl=openPipeline(j, false);
				for(String id:channelIds){
					pl.incrBy(buildCountInfoKeyByte(d, id, TaskCacheContext.SPACE_BYTELENTH), 0);
				}
				return syncAndReturnAllPipline(pl, false, Long.class, false);
			}
		});
		return ArrayUtils.toPrimitive(list.toArray(new Long[0]));
	}
	public long[] queryChannelAllByteLength(Date d,int taskType,String[] channelIds)throws Exception{
		return queryChannelAllByteLength(new CacheDataSpace(d, taskType),channelIds);
	};
	public Integer queryWaitTaskMaxPriority(final CacheDataSpace d,final String channelId)throws Exception{
		return excute(new TaskRedisCallback<Integer>() {

			@Override
			public Integer doInRedis(Jedis j) throws Exception {
				byte[] waitKey=buildChannelWaitQueueKeyByte(d,channelId);
				Set<byte[]> set=j.zrevrange(waitKey,0,0);
				if(set!=null && !set.isEmpty()){
					Double d=j.zscore(waitKey, set.iterator().next());
					return d==null?null:(int)d.doubleValue();
				}
				return null;
			}
		});
	};
	public Integer queryWaitTaskMaxPriority(Date d,int taskType,String channelId)throws Exception{
		return queryWaitTaskMaxPriority(new CacheDataSpace(d, taskType), channelId);
	}
	@Override
	public QueryResult query(final CacheDataSpace space, final QueryCondition condition,
			final int count) throws Exception {
		final QueryResult result=new QueryResult();
		if(count!=0 && condition!=null && !condition.getConditionMap().isEmpty()){
			excute(new TaskRedisCallback<Object>() {

				@Override
				public Object doInRedis(Jedis j) throws Exception {
					List<byte[]> keys=Lists.newArrayList(); 
					for(Entry<String, Object> en:condition.getConditionMap().entrySet()){
						if(en.getValue() instanceof String){
							keys.add(buildConditionSetKeyByte(space, en.getKey(), (String)en.getValue()));
						}
					}
					if(CollectionUtils.isNotEmpty(keys)){
						Set<byte[]> ids=j.sinter(buildBytess(keys));
						if(CollectionUtils.isNotEmpty(ids)){
							String resultId=UUID.randomUUID().toString();
							byte[] resultIdByte=RedisSerializeUtil.serializeString(
									buildQueryResultIdKey(resultId)
									);
							j.rpush(resultIdByte,buildBytess(ids));
							int expire=PropertyUtils.getIntProperty(
									TaskServicePropertiesLoader.TASKSERVICE_CACHE_QUERY_CACHERESULTEXIST_MAXEXPIRE, 
									TaskServicePropertiesLoader.FILE_NAME,120000);
							j.pexpire(resultIdByte, expire);//结果集做缓存
							result.setResultExpires(expire);
							List<byte[]> list=j.lrange(resultIdByte, 0, count-1);//分页取出
							result.setCount(count);
							if(CollectionUtils.isNotEmpty(list)){
								result.setResult(findCacheTasks(Sets.newLinkedHashSet(list), true, j));
								result.setResultId(resultId);
								result.setSum(ids.size());
							}
						}
					}
					return null;
				}
			});
		}
		return result;
	}
	@Override
	public QueryResult query(final String resultId, final int offset, final int count)
			throws Exception {
		final QueryResult result=new QueryResult();
		if(offset>=0 && count>0){
			excute(new TaskRedisCallback<Object>() {

				@Override
				public Object doInRedis(Jedis j) throws Exception {
					byte[] resultIdByte=RedisSerializeUtil.serializeString(
							buildQueryResultIdKey(resultId)
							);
					if(j.exists(resultIdByte) && "list".equals(j.type(resultIdByte))){
						List<byte[]> list=j.lrange(resultIdByte, offset, offset+count-1);//分页取出
						result.setOffset(offset);
						result.setCount(count);
						result.setResultExpires(j.pttl(resultIdByte));
						if(CollectionUtils.isNotEmpty(list)){
							result.setResult(findCacheTasks(Sets.newLinkedHashSet(list), true, j));
							result.setResultId(resultId);
							result.setSum((int)j.llen(resultIdByte).longValue());
						}
					}
					return null;
				}
			});
		}
		return result;
	}
	@Override
	public AbstractTaskModel query(String taskId) throws Exception {
		List<AbstractTaskModel> result=query(new String[]{taskId});
		return CollectionUtils.isNotEmpty(result)?result.get(0):null;
	}
	@Override
	public List<AbstractTaskModel> query(final String... taskIds) throws Exception {
		final List<AbstractTaskModel> result=Lists.newArrayList();
		if(org.apache.commons.lang3.ArrayUtils.isNotEmpty(taskIds)){
			excute(new TaskRedisCallback<Object>() {

				@Override
				public Object doInRedis(Jedis j) throws Exception {
					result.addAll(findCacheTasksByIds(Sets.newLinkedHashSet(Arrays.asList(taskIds)), j));
					return null;
				}
			});
		}
		return result;
	}

	@Override
	public List<byte[]> getKeys(String channelCode, String status) throws Exception {
		List<byte[]> keyList=new ArrayList<byte[]>();
		final String keys=KeyUtil.getKey(channelCode, status);;
		final Set<byte[]> keySets=Sets.newHashSet();
		excute(new TaskRedisCallback<Set<byte[]>>() {
			@Override
			public Set<byte[]> doInRedis(Jedis j) throws Exception {
				keySets.addAll(getKeysByKey(keys,j));
				return keySets;
			}
		});
		//转成List<String>格式
		if (keySets.size() > 0) {
			Iterator<byte[]>  it=keySets.iterator();
			while(it.hasNext()){
				keyList.add(it.next());
			}
		}
		return keyList;
	}

}
